package com.binance.assetservice.web.service.impl;

import java.io.IOException;
import java.math.BigDecimal;
import java.math.RoundingMode;
import java.util.Arrays;
import java.util.Date;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.regex.Matcher;
import java.util.regex.Pattern;
import java.util.stream.Collectors;

import javax.annotation.PostConstruct;

import com.binance.assetservice.web.configs.CustomRestTemplateConfig;
import org.apache.commons.collections4.CollectionUtils;
import org.apache.commons.collections4.MapUtils;
import org.apache.commons.lang3.BooleanUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.curator.shaded.com.google.common.collect.Sets;
import org.javasimon.SimonManager;
import org.javasimon.Split;
import org.javasimon.aop.Monitored;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Service;
import org.springframework.web.client.RestTemplate;
import org.springframework.web.util.UriComponentsBuilder;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.binance.assetservice.data.mapper.ProductItemMapper;
import com.binance.assetservice.data.models.ProductItem;
import com.binance.assetservice.vo.ProductItemVO;
import com.binance.assetservice.vo.request.product.FetchProductsRequest;
import com.binance.assetservice.vo.request.product.GetUserTradeFeeRequest;
import com.binance.assetservice.vo.request.product.PriceConvertRequest;
import com.binance.assetservice.vo.response.product.CurrencyResponse;
import com.binance.assetservice.vo.response.product.CurrencyResponse.PairRate;
import com.binance.assetservice.vo.response.product.FetchProductsResponse;
import com.binance.assetservice.vo.response.product.GetUserTradeFeeResponse;
import com.binance.assetservice.vo.response.product.PriceConvertResponse;
import com.binance.assetservice.web.cache.IProductSnapshotCache;
import com.binance.assetservice.web.service.IOrderBookManagerService;
import com.binance.assetservice.web.service.IProductService;
import com.binance.assetservice.web.utils.Graph;
import com.binance.autojob.annotation.MyScheduledJob;
import com.binance.master.enums.SymbolStatus;
import com.binance.master.error.BusinessException;
import com.binance.master.error.GeneralCode;
import com.binance.master.utils.CopyBeanUtils;
import com.binance.master.utils.DateUtils;
import com.binance.master.utils.FormatUtils;
import com.binance.master.utils.RedisCacheUtils;
import com.binance.sysconf.service.SysConfigVarCacheService;
import com.google.common.cache.CacheBuilder;
import com.google.common.cache.CacheLoader;
import com.google.common.cache.LoadingCache;
import com.google.common.cache.RemovalListener;
import com.google.common.cache.RemovalNotification;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import com.rabbitmq.client.Address;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;

import lombok.extern.slf4j.Slf4j;

@Service
@Slf4j
public class ProductServiceImpl extends BaseServiceImpl implements IProductService {
    // private static final String PRODUCT_ITEM_BY_SYMBOL_CACHE_KEY_PREFIX = "product_item.symbol";
    // private static final String USER_TRADE_FEE_CACHE_KEY = "user.trade.fee";
    private static List<String> currencyList = Lists.newArrayList();
    private static List<String> defaultCurrencyList = Lists.newArrayList();
    static {
        defaultCurrencyList.add("CNY_USD");
        defaultCurrencyList.add("EUR_USD");
        defaultCurrencyList.add("KRW_USD");
        defaultCurrencyList.add("JPY_USD");
        defaultCurrencyList.add("GBP_USD");
        defaultCurrencyList.add("RUB_USD");
    }
    final public static long NINE_O_CLOCK = 60 * 60 * 1000 * 9;
    private final String PERFORMANCE_METRICS_NAME = this.getClass().getName() + ".fetchProducts";
    @Autowired
    private ProductItemMapper productItemMapper;
    @Autowired
    private SysConfigVarCacheService sysConfigService;
    @Autowired
    private IProductSnapshotCache productSnapshotCache;
    @Autowired
    private IOrderBookManagerService orderBookManager;
    @Autowired
    private RestTemplate restTemplate;
    @Autowired
    private CustomRestTemplateConfig.CustomRestTemplate customRestTemplate;
    @Autowired
    private RedisTemplate<String, Object> redisTemplate;
    @Value("${matchbox.rest.root.url}")
    private String restUrl;
    @Value("${matching.unit.type}")
    private String matchingType;
    @Value("${spring.rabbitmq.username}")
    private String rabbitmqUsername;
    @Value("${spring.rabbitmq.password}")
    private String rabbitmqPassword;
    @Value("${spring.rabbitmq.addresses}")
    private String rabbitmqAddresses;
    @Value("${op.juhe.cn.key:75be1def356f2bd62579491e45b013f1}")
    private String opJuheKey;
    @Value("${currency.pair:}")
    private String currencyPair;
    private List<String> visibleSymbols;
    private Graph productGraph = new Graph();
    private Double cnyusd = 6.6;
    private Map<String, BigDecimal> currencyMap = Maps.newHashMap();

    private LoadingCache<String, ProductItem> productItemLoadingCache =
            CacheBuilder.newBuilder().maximumSize(10000).removalListener(new RemovalListener<String, ProductItem>() {

                @Override
                public void onRemoval(RemovalNotification<String, ProductItem> notification) {}

            }).build(new CacheLoader<String, ProductItem>() {

                @Override
                public ProductItem load(String key) throws Exception {
                    return productItemMapper.getProductBySymbol(key);
                }

            });

    private LoadingCache<String, Map<String, Object>> productDataLoadingCache =
            CacheBuilder.newBuilder().maximumSize(10000).removalListener(notification -> {
            }).build(new CacheLoader<String, Map<String, Object>>() {

                @Override
                public Map<String, Object> load(String key) throws Exception {
                    ProductItem productItem = productItemLoadingCache.get(key);
                    if (null == productItem) {
                        log.warn("Can't find symbol: {} in cache", key);
                        return null;
                    }
                    String symbol = productItem.getSymbol();
                    Map<String, Object> obj = Maps.newHashMap();
                    obj.put("symbol", symbol);
                    obj.put("baseAsset", productItem.getBaseAsset());
                    obj.put("quoteAsset", productItem.getQuoteAsset());
                    obj.put("baseAssetUnit", productItem.getBaseAssetUnit());
                    obj.put("quoteAssetUnit", productItem.getQuoteAssetUnit());
                    obj.put("baseAssetName", productItem.getBaseAssetName());
                    obj.put("quoteAssetName", productItem.getQuoteAssetName());
                    obj.put("desc", productItem.getDescription());
                    obj.put("active", productItem.getActive());
                    obj.put("tickSize", FormatUtils.getPriceNumericFormatter().format(productItem.getTickSize()));
                    obj.put("stepSize", FormatUtils.getPriceNumericFormatter().format(productItem.getStepSize()));
                    obj.put("minTrade", productItem.getMinTrade());
                    obj.put("market", productItem.getMarket());
                    obj.put("marketName", productItem.getMarketName());
                    obj.put("parentMarket", productItem.getParentMarket());
                    obj.put("parentMarketName", productItem.getParentMarketName());
                    if (sysConfigService.getValue("product_withdraw_fee") != null) {
                        obj.put("withdrawFee", sysConfigService.getValue("product_withdraw_fee"));
                    }
                    /// --------
                    // 前一天的
                    Boolean productSnapshotFlag = productSnapshotCache.getProductSnapshotFlag(symbol);
                    double basePrice = 0D;
                    if (productItem.getClosePrice() != null) {
                        if (productSnapshotFlag != null && productSnapshotFlag && productItem.getPrevClose() != null) {
                            basePrice = productItem.getPrevClose().doubleValue();
                        } else {
                            basePrice = productItem.getClosePrice().doubleValue();
                        }
                    }
                    obj.put("prevClose", basePrice);
                    obj.put("status", productItem.getStatus());
                    obj.put("minQty", productItem.getMinQty());
                    obj.put("decimalPlaces", productItem.getDecimalPlaces());
                    obj.put("matchingUnitType", productItem.getMatchingUnitType());
                    return obj;
                }

            });

    private LoadingCache<String, Map<String, Object>> userTradeFeeLoadingCache =
            CacheBuilder.newBuilder().maximumSize(10000).expireAfterAccess(60, TimeUnit.MINUTES)
                    .removalListener(new RemovalListener<String, Map<String, Object>>() {

                        @Override
                        public void onRemoval(RemovalNotification<String, Map<String, Object>> notification) {}

                    }).build(new CacheLoader<String, Map<String, Object>>() {

                        @Override
                        public Map<String, Object> load(String key) throws Exception {
                            String[] arr = StringUtils.split(key, "#");
                            return productItemMapper.getUserTradeFee(arr[1], arr[0]);
                        }

                    });

    @PostConstruct
    public void init() throws Exception {
        this.refreshRates();
        this.refreshProductAll(false);
    }

    public void refreshCurrencyPair() {
        currencyList.clear();
        if (StringUtils.isNotBlank(currencyPair)) {
            currencyList.addAll(Arrays.asList(StringUtils.split(currencyPair, ",")));
        } else {
            currencyList.addAll(defaultCurrencyList);
        }
    }

    // @Scheduled(initialDelay = 5 * 60 * 1000, fixedDelay = 5 * 60 * 1000)
    @MyScheduledJob(cron = "0 0/30 * * * ?", author = "wangxiao", alarmEmail = "louis.wang@binance.com",
            jobDesc = "每隔30分钟执行一次刷新汇率")
    public void refreshRates() {
        refreshCurrencyPair();
        refreshCNYUSD();
        refreshCurrency();
    }

    private void refreshCNYUSD() {
        try {
            String url = "http://op.juhe.cn/onebox/exchange/query?key=" + opJuheKey;
            UriComponentsBuilder builder = UriComponentsBuilder.fromHttpUrl(url);
            log.info("刷新汇率: {}", builder.build().toUriString());
            String rate = this.customRestTemplate.getRestTemplate().postForObject(builder.build().toUri(), null,
                    String.class);
            JSONObject object = JSON.parseObject(rate);
            if (object.getString("reason").contains("查询成功") || object.getString("reason").contains("successed")) {
                JSONArray usd = null;
                JSONArray list = object.getJSONObject("result").getJSONArray("list");

                for (int i = 0; i < list.size(); i++) {
                    JSONArray current = list.getJSONArray(i);
                    if (StringUtils.equals(current.getString(0), "美元")) {
                        usd = current;
                        break;
                    }
                }
                Double cnyusd = usd.getDouble(2) / usd.getDouble(1);
                RedisCacheUtils.set("CNY_USD", cnyusd, 0, "exchange:rate");
            } else {
                log.warn("刷新汇率异常: {}", rate);
            }
        } catch (Exception e) {
            log.error("刷新汇率异常: {}", e);
        }
    }

    private void getCurrencyRate(String currencyPair, StringBuilder exceptionSb) {
        // 对于每个交易对CNY_USD，查询汇率
        String[] splitStr = StringUtils.split(currencyPair, "_");
        if (splitStr.length < 2) {
            return;
        }
        try {
            String url = "http://op.juhe.cn/onebox/exchange/currency?key=" + opJuheKey + "&from=" + splitStr[0] + "&to="
                    + splitStr[1];
            UriComponentsBuilder builder = UriComponentsBuilder.fromHttpUrl(url);
            log.info("刷新汇率: {}", builder.build().toUriString());
            String rate = this.customRestTemplate.getRestTemplate().postForObject(builder.build().toUri(), null,
                    String.class);
            JSONObject object = JSON.parseObject(rate);

            if (object.getString("reason").contains("查询成功") || object.getString("reason").contains("successed")) {
                JSONArray currencyList = object.getJSONArray("result");
                if (currencyList != null && currencyList.size() > 0) {
                    JSONObject toCurrency = (JSONObject) currencyList.get(1);
                    double rate1 = toCurrency.getDouble("result");
                    double rate2 = 0d;
                    if (StringUtils.isBlank(toCurrency.getString("currencyF_Name"))) {
                        // 如果currencyF_Name为空，一般是缺少数据，尝试第二个数据源
                        rate2 = getRateFromCurrencyConverter(splitStr[0], splitStr[1]);
                    }
                    currencyMap.put(currencyPair, BigDecimal.valueOf(rate2 > 0 ? rate2 : rate1));
                }
            } else {// 错误则尝试调用第二个数据源
                double secondRate = getRateFromCurrencyConverter(splitStr[0], splitStr[1]);
                if (secondRate > 0) {// 正确调用并且返回数据
                    currencyMap.put(currencyPair, BigDecimal.valueOf(secondRate));
                }
            }
        } catch (Exception e) {
            exceptionSb.append(e.getMessage());
            exceptionSb.append("\n");
            log.error("刷新汇率异常: {}---开始尝试第二个数据源", e);
            double secondRate = getRateFromCurrencyConverter(splitStr[0], splitStr[1]);
            if (secondRate > 0) {// 正确调用并且返回数据
                currencyMap.put(currencyPair, BigDecimal.valueOf(secondRate));
            }
        }
    }

    private void refreshCurrency() {
        try {
            StringBuilder exceptionSb = new StringBuilder();
            for (String currencyPair : currencyList) {
                this.getCurrencyRate(currencyPair, exceptionSb);
            }
            if (exceptionSb.length() > 0) {
                log.error("刷新汇率异常: {}", exceptionSb.toString());
            }
        } catch (Exception e) {
            log.error("刷新汇率异常: {}", e);
        }

        try {
            Map<String, BigDecimal> cachedMap = RedisCacheUtils.get("currency", Map.class, "exchange:all:rate");
            if (cachedMap != null) {
                if (!currencyMap.isEmpty()) {
                    for (Map.Entry<String, BigDecimal> entry : currencyMap.entrySet()) {
                        cachedMap.put(entry.getKey(), entry.getValue());
                    }
                    RedisCacheUtils.set("currency", cachedMap, 0, "exchange:all:rate");
                }
            } else {
                RedisCacheUtils.set("currency", currencyMap, 0, "exchange:all:rate");
            }
        } catch (Exception e) {
            log.error("刷新redis中汇率信息出现异常: {}", e);
        }

    }

    private double getRateFromCurrencyConverter(String fromCurrency, String toCurrency) {
        double rtnVal = 0d;
        if (!StringUtils.isAnyBlank(fromCurrency, toCurrency)) {
            try {
                // reverse it
                String currencyPair = toCurrency + "_" + fromCurrency;
                String url = String.format("http://free.currencyconverterapi.com/api/v5/convert?q=%s&compact=y",
                        currencyPair);
                UriComponentsBuilder builder = UriComponentsBuilder.fromHttpUrl(url);
                log.info("刷新汇率: {}", builder.build().toUriString());
                String rate = this.restTemplate.getForObject(builder.build().toUri(), String.class);
                JSONObject object = JSON.parseObject(rate);
                rtnVal = object.getJSONObject(currencyPair).getDouble("val");
            } catch (Exception ex) {
                log.error("刷新汇率失败：{}", ex);
            }
        }
        return rtnVal;
    }

    @Monitored
    @Override
    public List<ProductItem> getAllProducts() {
        return productItemMapper.fetchAllProducts();
    }

    @Monitored
    @Override
    public List<String> getVisibleSymbols() {
        return visibleSymbols;
    }

    @Override
    public ProductItem get(String symbol) {
        this.validateSymbol(Lists.newArrayList(symbol));
        try {
            return this.productItemLoadingCache.get(symbol);
        } catch (Exception e) {
            log.warn("根据产品编码{}查询缓存数据出错：{}", symbol, e);
            throw new RuntimeException(e);
        }
    }


    @Monitored
    @Override
    public List<ProductItem> getBySymbols(String[] symbols) {
        // Map<String, ProductItem> map = this.getBySymbolAsMap(symbols);
        // return Lists.newArrayList(map.values().iterator());
        List<ProductItem> list = Lists.newArrayList();
        if (symbols != null) {
            Set<String> symbolSet = this.validateSymbol(Lists.newArrayList(symbols));
            for (String symbol : symbolSet) {
                try {
                    ProductItem pi = this.productItemLoadingCache.get(symbol);
                    if (pi != null) {
                        list.add(pi);
                    }
                } catch (Exception e) {
                    log.error("根据产品编码{}查询缓存数据出错：{}", symbol, e);
                }
            }
        }
        return list;
    }

    @Override
    public void addProduct(ProductItem product) {
        this.productItemMapper.addProduct(product);
        this.refreshProductAll();
    }

    @Override
    public void updateProduct(ProductItemVO product) {
        if (product == null) {
            log.error("更新产品为空");
            return;
        }
        int rt = this.productItemMapper.updateProduct(CopyBeanUtils.copy(product, ProductItem.class));
        if (rt > 0) {
            refreshProduct(product.getSymbol(), true);
        }
    }

    private Set<String> validateSymbol(List<String> symbols) {
        if (CollectionUtils.isEmpty(symbols)) {
            throw new IllegalArgumentException("产品编码不能为空");
        }
        for (String symbol : symbols) {
            // 长度不得超过20个
            if (symbol.length() > 20) {
                throw new IllegalArgumentException("产品编码不能超过20个字符长度");
            }
            // 格式必须数字加字符
            Pattern pattern = Pattern.compile("^[A-Za-z0-9]+$");
            Matcher m = pattern.matcher(symbol);
            if (!m.find()) {
                throw new IllegalArgumentException("产品编码必须匹配：^[A-Za-z0-9]+$");
            }
        }
        return Sets.newHashSet(symbols.iterator());
    }

    @Override
    public Double getSumCirculationBySymbol(String symbol) {
        Map<String, Object> pd = this.getProductData(symbol);
        if (pd == null) {
            return null;
        } else {
            return (Double) pd.get("circulation");
        }
    }

    @Override
    public Double getSumTotalBySymbol(String symbol) {
        Map<String, Object> pd = this.getProductData(symbol);
        if (pd == null) {
            return null;
        } else {
            return (Double) pd.get("total");
        }
    }

    @Override
    public BigDecimal getPrevClose(String symbol) {
        Map<String, Object> pd = this.getProductData(symbol);
        if (pd == null) {
            return null;
        } else {
            Object prevClose = pd.get("prevClose");
            if (prevClose instanceof BigDecimal) {
                return (BigDecimal) prevClose;
            } else if (prevClose instanceof Double) {
                return BigDecimal.valueOf((Double) prevClose);
            } else {
                log.error("数据错误{}", symbol);
                return (BigDecimal) prevClose;
            }
        }
    }

    @Override
    public void refreshProduct(String symbol, boolean refreshOtherServers) {
        if (refreshOtherServers) {
            this.sendMQ2RefreshProduct(symbol);
        } else {
            this.productItemLoadingCache.invalidate(symbol);
            this.get(symbol);

            this.productDataLoadingCache.invalidate(symbol);
            this.getProductData(symbol);
        }
    }

    // 9点 刷新全部产品列表
    // @Scheduled(cron = "0 0 9 * * ?")
    @MyScheduledJob(cron = "0 0 9 * * ?", author = "wangxiao", alarmEmail = "louis.wang@binance.com",
            jobDesc = "每天9点刷新全部产品列表")
    public void refreshProductAll() {
        this.refreshProductAll(true);
    }

    @Override
    public void refreshProductAll(boolean refreshOtherServers) {
        if (refreshOtherServers) {
            this.sendMQ2RefreshProduct(null);
        } else {
            for (String symbol : this.productItemMapper.getAllSymbols()) {
                this.refreshProduct(symbol, false);
            }
            visibleSymbols = this.productItemMapper.getVisibleSymbols();
            if (visibleSymbols == null) {
                visibleSymbols = Lists.newArrayList();
            }
            try {
                this.orderBookManager.reInit();
            } catch (Exception e) {
                log.error("刷新OrderBook出现异常：", e);
            }
            Set<String> visibleSet = new HashSet<String>(visibleSymbols);
            productGraph.clear();
            for (ProductItem product : getAllProducts()) {
                if (visibleSet.contains(product.getSymbol())
                        && (StringUtils.equals(product.getStatus(), SymbolStatus.TRADING.toString())
                                || StringUtils.equals(product.getStatus(), SymbolStatus.BREAK.toString()))) {
                    productGraph.addEdge(product.getBaseAsset(), product.getQuoteAsset(), product.getSymbol());
                }
                productSnapshotCache.setProductSnapshotFlag(product.getSymbol(), false);
            }
        }
    }

    @Override
    public Map<String, Object> getProductData(String productSymbol) {
        try {
            return this.productDataLoadingCache.get(productSymbol);
        } catch (ExecutionException e) {
            log.warn("根据产品编码{}查询缓存数据ProductData出错：{}", productSymbol, e);
            throw new RuntimeException(e);
        }

    }

    @Override
    public Double getFirstPrevClose(String symbol) {
        return this.productItemMapper.getFirstPrevClose(symbol);
    }

    @Override
    public Double getCnyusd() {
        try {
            return RedisCacheUtils.get("CNY_USD", Double.class, "exchange:rate");
        } catch (Exception e) {
            log.warn("failed to get CNY_USD:{}", e);
            return cnyusd;
        }
    }

    @SuppressWarnings("unchecked")
    @Override
    public CurrencyResponse getCurrency() {
        Map<String, BigDecimal> map = null;
        CurrencyResponse response = new CurrencyResponse();
        try {
            map = RedisCacheUtils.get("currency", Map.class, "exchange:all:rate");
        } catch (Exception e) {
            log.warn("failed to get exchangeAllRates:{}", e);
            map = currencyMap;
        }
        if (map != null) {
            map.forEach((key, value) -> {
                PairRate pr = new PairRate();
                pr.setPair(key);
                pr.setRate(value);
                response.getPairRates().add(pr);
            });
        }
        return response;
    }

    @Override
    public Map<String, Object> getUserTradeFee(Map<String, Object> param) {
        String userId = param.get("userId").toString();
        String symbol = param.get("symbol").toString();
        String key = userId + "#" + symbol;
        try {
            return this.userTradeFeeLoadingCache.get(key);
        } catch (ExecutionException e) {
            log.warn("根据Key({})没有从缓存中查询到数据：{}", key, e);
            throw new RuntimeException(e);
        }
    }

    @Override
    public void refreshUserTradeFee(Long userId, String symbol, boolean refreshOtherServers) {
        if (refreshOtherServers) {
            this.sendMQ2RefreshUserProductTradeFee(userId, symbol);
        } else {
            String key = userId + "#" + symbol;
            this.userTradeFeeLoadingCache.invalidate(key);
        }
    }

    @Override
    public void invalidateAllUserTradeFee(boolean refreshOtherServers) {
        if (refreshOtherServers) {
            this.sendMQ2RefreshUserProductTradeFee(null, null);
        } else {
            this.userTradeFeeLoadingCache.invalidateAll();
        }
    }

    @Override
    public List<String> getShortestPath(String from, String to) {
        return productGraph.getShortestPathEdge(from, to);
    }

    @Override
    @Monitored
    public List<FetchProductsResponse> fetchProducts(FetchProductsRequest body) throws Exception {
        List<FetchProductsResponse> responseList = Lists.newArrayList();
        log.debug("Received product fetch request, body: {}", body);
        List<String> list = Lists.newLinkedList();
        if (StringUtils.isBlank(body.getSymbol())) {
            list.addAll(this.getVisibleSymbols());
        } else {
            if (StringUtils.isNotBlank(body.getSymbol())) {
                String[] symbols = body.getSymbol().split(",");// 支持批量查询产品信息，兼容老接口
                if (symbols != null && symbols.length > 0) {
                    List<ProductItem> productItems = this.getBySymbols(symbols);
                    if (CollectionUtils.isEmpty(productItems)) {
                        log.warn("Can't find symbol: {} in cache", StringUtils.join(symbols, ","));
                    } else {
                        list = productItems.stream().filter(singleProductItem -> singleProductItem.getTest() != 1)
                                .map(singleProductItem -> singleProductItem.getSymbol()).collect(Collectors.toList());
                    }
                }
            }
        }
        if (CollectionUtils.isNotEmpty(list)) {
            responseList.addAll(this.transformer(list, BooleanUtils.isTrue(body.getIsWidget())));
        }
        return responseList;
    }

    private List<FetchProductsResponse> transformer(List<String> symbols, boolean isWidget) {
        List<FetchProductsResponse> responseList = Lists.newArrayList();
        for (String symbol : symbols) {
            try {
                Map<String, Object> productData = this.getProductData(symbol);
                if (productData != null) {
                    FetchProductsResponse response = CopyBeanUtils.copy(productData, FetchProductsResponse.class);
                    ProductItem product = this.get(symbol);
                    int stepSize = 8;
                    int tickSize = 8;
                    if (product != null) {
                        stepSize = product.getTickSize().stripTrailingZeros().scale();
                        tickSize = product.getTickSize().stripTrailingZeros().scale();
                    }
                    response.setOpen(new BigDecimal(String.valueOf(orderBookManager.getOpen(symbol))).setScale(tickSize,
                            BigDecimal.ROUND_CEILING));
                    response.setHigh(new BigDecimal(String.valueOf(orderBookManager.getHigh(symbol))).setScale(tickSize,
                            BigDecimal.ROUND_CEILING));// 当天最高
                    response.setLow(new BigDecimal(String.valueOf(orderBookManager.getLow(symbol))).setScale(tickSize,
                            BigDecimal.ROUND_CEILING));// 当天最低

                    BigDecimal close = new BigDecimal(String.valueOf(orderBookManager.getClose(symbol)))
                            .setScale(tickSize, BigDecimal.ROUND_CEILING);
                    response.setClose(close);// 最新价
                    response.setVolume(new BigDecimal(String.valueOf(orderBookManager.getVolume(symbol)))
                            .setScale(stepSize, BigDecimal.ROUND_CEILING));// 累计
                    response.setTradedMoney(BigDecimal.valueOf(orderBookManager.getTurnOver(symbol)));
                    response.setActiveBuy(BigDecimal.valueOf(orderBookManager.getActiveBuy(symbol)));
                    response.setActiveSell(BigDecimal.valueOf(orderBookManager.getActiveSell(symbol)));
                    if (isWidget) {
                        // 最新价折算成美元价格
                        String quoteAsset = response.getQuoteAsset();
                        BigDecimal closeUSDollar = "USDT".equalsIgnoreCase(quoteAsset) ? close
                                : calcCloseToUSDollar(quoteAsset, close, tickSize);
                        response.setCloseUSDollar(closeUSDollar.setScale(2, BigDecimal.ROUND_HALF_UP));
                        // 最新价折算成人民币
                        response.setCloseRMB(calcCloseToRMB(closeUSDollar).setScale(2, BigDecimal.ROUND_HALF_UP));
                    }
                    responseList.add(response);
                }
            } catch (Exception e) {
                log.error("调用fetchProducts出现异常：", e);
            }
        }
        return responseList;
    }

    private BigDecimal calcCloseToUSDollar(String quoteAsset, BigDecimal originClose, int tickSize) {
        String symbol = quoteAsset + "USDT";
        log.info("calcCloseToUSDollar:{},the originClose is {}", symbol, originClose);
        return originClose.multiply(new BigDecimal(String.valueOf(orderBookManager.getClose(symbol))).setScale(tickSize,
                BigDecimal.ROUND_CEILING));
    }

    private BigDecimal calcCloseToRMB(BigDecimal originalUSDollar) {
        return originalUSDollar.multiply(new BigDecimal(this.getCnyusd()));
    }

    @Override
    @Monitored
    public GetUserTradeFeeResponse getUserTradeFee(GetUserTradeFeeRequest body) throws Exception {
        GetUserTradeFeeResponse response = new GetUserTradeFeeResponse();
        if (StringUtils.isBlank(body.getSymbol()) || StringUtils.isBlank(body.getUserId())
                || this.get(body.getSymbol()) == null) {
            response.setBuyMakerCommission("--");
            response.setBuyTakerCommission("--");
            response.setSellMakerCommission("--");
            response.setSellTakerCommission("--");
        } else {
            Map<String, Object> param = Maps.newHashMap();
            param.put("userId", body.getUserId());
            param.put("symbol", body.getSymbol());
            Map<String, Object> result = this.getUserTradeFee(param);
            if (MapUtils.isEmpty(result)) {
                response.setBuyMakerCommission("--");
                response.setBuyTakerCommission("--");
                response.setSellMakerCommission("--");
                response.setSellTakerCommission("--");
            } else {
                response.setBuyMakerCommission(
                        ((BigDecimal) result.get("buyMakerCommission")).compareTo(BigDecimal.ZERO) < 0 ? "0"
                                : ((BigDecimal) result.get("buyMakerCommission")).toPlainString());
                response.setBuyTakerCommission(
                        ((BigDecimal) result.get("buyTakerCommission")).compareTo(BigDecimal.ZERO) < 0 ? "0"
                                : ((BigDecimal) result.get("buyTakerCommission")).toPlainString());
                response.setSellMakerCommission(
                        ((BigDecimal) result.get("sellMakerCommission")).compareTo(BigDecimal.ZERO) < 0 ? "0"
                                : ((BigDecimal) result.get("sellMakerCommission")).toPlainString());
                response.setSellTakerCommission(
                        ((BigDecimal) result.get("sellTakerCommission")).compareTo(BigDecimal.ZERO) < 0 ? "0"
                                : ((BigDecimal) result.get("sellTakerCommission")).toPlainString());
            }
        }
        return response;
    }

    @Override
    @Monitored
    public PriceConvertResponse priceConvert(PriceConvertRequest body) throws Exception {
        Split split = SimonManager.getStopwatch(PERFORMANCE_METRICS_NAME).start();
        PriceConvertResponse response = new PriceConvertResponse();
        try {
            if (body.getFrom().equals(body.getTo())) {
                response.setPrice(BigDecimal.ONE);
                response.setAmount(body.getAmount());
            } else {
                BigDecimal price = getPrice(body.getFrom(), body.getTo(), body.getDate());
                if (price == null) {
                    log.error("failed to convert to : price is null");
                    throw new BusinessException(GeneralCode.ERROR, "price is null");
                } else {
                    response.setPrice(price);
                    response.setAmount(price.multiply(body.getAmount()));
                }
            }
        } finally {
            split.stop();
        }
        return response;
    }

    @Override
    public BigDecimal getPrice(String from, String to, Date date) {
        log.info("ProductServiceImpl.getPrice {},{}", from, to);
        BigDecimal price = new BigDecimal(1);
        // 法币系统由于没有ethbtc交易对，所以从主站获取最新价
        String restItem = sysConfigService.getValue("matchbox_external_rest_url");
        String productsItem = sysConfigService.getValue("external_products_list");
        if (StringUtils.isNotBlank(productsItem)) {
            String[] productList = StringUtils.split(productsItem, ",");
            boolean isMatch = false;
            for (String symbol : productList) {
                if ((from + to).equalsIgnoreCase(symbol)) {
                    isMatch = true;
                    break;
                }
            }
            if (StringUtils.isNotBlank(restItem) && isMatch) {
                return this.getLastPrice((from + to).toUpperCase(), restItem, price);
            }
        }
        List<String> path = this.getShortestPath(from, to);
        if (path == null || path.size() == 0) {
            log.warn("ProductServiceImpl.getPrice 没有获取到最短路径，返回Null：{},{}", from, to);
            return null;
        }
        log.info("ProductServiceImpl.getPrice {},{}获取到最短路径{}", from, to, path);
        String asset = from;
        for (String symbol : path) {
            ProductItem product = this.get(symbol);
            if (product.getTest() != null && product.getTest() == 1) {
                log.warn("ProductServiceImpl.getPrice 是测试产品，返回0。{},{}", from, to);
                return BigDecimal.ZERO;
            }
            if (product.getBaseAsset().equals(asset)) {
                if (date == null) {
                    Double close = orderBookManager.getClose(symbol);
                    log.info("ProductServiceImpl.getPrice {},the close is {}", close);
                    if (close == null || close == 0) {
                        JSONArray result = this.fetchLatestKline(null, symbol);
                        if (result == null || result.isEmpty()) {
                            price = new BigDecimal(0); // 不存在产品 返回 0
                        } else {
                            price = price.multiply(new BigDecimal(result.getString(4)));
                        }
                    } else {
                        price = price.multiply(new BigDecimal(String.valueOf(close)));
                    }
                } else {
                    JSONArray result = this.fetchKlineOn(symbol, date);
                    if (result == null || result.isEmpty()) {
                        price = new BigDecimal(0); // 不存在产品 返回 0
                    } else {
                        price = price.multiply(new BigDecimal(result.getString(4)));
                    }
                }
                asset = product.getQuoteAsset();
            } else {
                if (date == null) {
                    Double close = orderBookManager.getClose(symbol);
                    log.info("ProductServiceImpl.getPrice {},the close is {}", close);
                    if (close == null || close == 0) {
                        JSONArray result = this.fetchLatestKline(null, symbol);
                        if (result == null || result.isEmpty()) {
                            price = new BigDecimal(0); // 不存在产品 返回 0
                        } else {
                            price = price.divide(new BigDecimal(result.getString(4)), 8, RoundingMode.DOWN);
                        }
                    } else {
                        price = price.divide(new BigDecimal(String.valueOf(close)), 8, RoundingMode.DOWN);
                    }
                } else {
                    JSONArray result = this.fetchKlineOn(symbol, date);
                    if (result == null || result.isEmpty()) {
                        price = new BigDecimal(0); // 不存在产品 返回 0
                    } else {
                        price = price.divide(new BigDecimal(result.getString(4)), 8, RoundingMode.DOWN);
                    }
                }
                asset = product.getBaseAsset();
            }
        }
        return price;
    }

    private BigDecimal getLastPrice(String symbol, String restUrl, BigDecimal price) {
        Double close = orderBookManager.getClose(symbol);
        log.info("ProductServiceImpl.getLastPrice {},{}", symbol, close);
        if (close == null || close.compareTo(0d) <= 0) {
            JSONArray result = this.fetchLatestKline(restUrl, symbol);
            log.info("ProductServiceImpl.getLastPrice.fetchLatestKline {},{},{}", restUrl, symbol, result);
            if (result == null) {
                price = new BigDecimal(0); // 不存在产品 返回 0
            } else {
                price = price.multiply(new BigDecimal(result.getString(4)));
            }
        } else {
            price = price.multiply(new BigDecimal(String.valueOf(close)));
        }
        return price;
    }

    private JSONArray fetchLatestKline(String restUrl, String symbol) {
        String url = UriComponentsBuilder
                .fromHttpUrl(String.format("%s/klines", StringUtils.isBlank(restUrl) ? this.restUrl : restUrl))
                .queryParam("symbol", symbol).queryParam("limit", "1").queryParam("interval", "1d").build()
                .toUriString();
        String jsonStr = this.restTemplate.getForObject(url, String.class);
        log.info("fetchLatestKline:{} the result is {}", symbol, jsonStr);
        JSONArray array = JSON.parseArray(jsonStr);
        if (array.size() > 0) {
            return array.getJSONArray(0);
        }
        return null;
    }

    private JSONArray fetchKlineOn(String symbol, Date date) {
        long startTime = date.getTime();
        long endTime = startTime + 24 * 60 * 60 * 1000 - 1;
        String url = UriComponentsBuilder.fromHttpUrl(String.format("%s/klines", restUrl)).queryParam("symbol", symbol)
                .queryParam("limit", "2").queryParam("interval", "1d").queryParam("startTime", startTime)
                .queryParam("endTime", endTime).build().toUriString();
        String jsonStr = this.restTemplate.getForObject(url, String.class);
        log.info("fetchLatestKline:{},{} the result is {}", symbol, date, jsonStr);
        JSONArray array = JSON.parseArray(jsonStr);
        if (array.size() > 0) {
            return array.getJSONArray(0);
        }
        return null;
    }

    @Override
    public List<ProductItem> getProductByRuleId(Long ruleId) throws Exception {
        return this.productItemMapper.getProductByRuleId(ruleId);
    }

    private void sendMQ2RefreshProduct(String symbol) {
        String exchangeName = "product_refresh";
        String message = DateUtils.getNewUTCDate().getTime() + " product refresh...";
        if (StringUtils.isNotBlank(symbol)) {
            message += "#" + symbol;
        }
        ConnectionFactory factory = new ConnectionFactory();
        // MQ集群模式
        factory.setAutomaticRecoveryEnabled(true);
        factory.setUsername(this.rabbitmqUsername);
        factory.setPassword(this.rabbitmqPassword);
        Connection conn = null;
        Channel channel = null;
        try {
            // 创建connection
            conn = factory.newConnection(Address.parseAddresses(this.rabbitmqAddresses));
            // 创建channel
            channel = conn.createChannel();
            // 声明该channel是fanout类型
            channel.exchangeDeclare(exchangeName, "fanout");
            // 将消息发送给exchange
            channel.basicPublish(exchangeName, "", null, message.getBytes());
        } catch (Exception e) {
            log.error("向mq发送信息通知各个服务节点刷新缓存", e);
        } finally {
            if (channel != null) {
                try {
                    channel.close();
                } catch (IOException | TimeoutException e) {
                }
            }
            if (conn != null) {
                try {
                    conn.close();
                } catch (IOException e) {
                }
            }
        }
    }

    private void sendMQ2RefreshUserProductTradeFee(Long userId, String symbol) {
        String exchangeName = "user_ProductTradeFee_refresh";
        String message = DateUtils.getNewUTCDate().getTime() + " UserProductTradeFee refresh...";
        if (StringUtils.isNotBlank(symbol) && userId != null) {
            message = userId + "#" + symbol;
        }
        ConnectionFactory factory = new ConnectionFactory();
        // MQ集群模式
        factory.setAutomaticRecoveryEnabled(true);
        factory.setUsername(this.rabbitmqUsername);
        factory.setPassword(this.rabbitmqPassword);
        Connection conn = null;
        Channel channel = null;
        try {
            // 创建connection
            conn = factory.newConnection(Address.parseAddresses(this.rabbitmqAddresses));
            // 创建channel
            channel = conn.createChannel();
            // 声明该channel是fanout类型
            channel.exchangeDeclare(exchangeName, "fanout");
            // 将消息发送给exchange
            channel.basicPublish(exchangeName, "", null, message.getBytes());
        } catch (Exception e) {
            log.error("向mq发送信息通知各个服务节点刷新缓存", e);
        } finally {
            if (channel != null) {
                try {
                    channel.close();
                } catch (IOException | TimeoutException e) {
                }
            }
            if (conn != null) {
                try {
                    conn.close();
                } catch (IOException e) {
                }
            }
        }
    }

}
